package com.example.rocketmq.demo.consumer;

import com.example.rocketmq.demo.config.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 高级消费者
 * 展示不同的消费模式和配置
 */
@Slf4j
@Component
public class AdvancedConsumer {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    private DefaultMQPushConsumer broadcastConsumer;
    private DefaultMQPushConsumer clusteringConsumer;
    private DefaultMQPushConsumer retryConsumer;

    @PostConstruct
    public void init() throws MQClientException {
        // 1. 广播模式消费者
        initBroadcastConsumer();

        // 2. 集群模式消费者
        initClusteringConsumer();

        // 3. 重试机制消费者
        initRetryConsumer();
    }

    /**
     * 初始化广播模式消费者
     * 广播模式下，每个消费者实例都会收到所有消息
     */
    private void initBroadcastConsumer() throws MQClientException {
        broadcastConsumer = new DefaultMQPushConsumer(RocketMQConstant.CONSUMER_GROUP_NORMAL + "-broadcast");
        broadcastConsumer.setNamesrvAddr(nameServer);
        broadcastConsumer.subscribe(RocketMQConstant.TOPIC_NORMAL, "*");

        // 设置为广播模式
        broadcastConsumer.setMessageModel(MessageModel.BROADCASTING);

        // 从最新的消息开始消费
        broadcastConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        broadcastConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                log.info("广播模式消费者接收到消息: msgId={}, topic={}, tags={}, content={}",
                        msg.getMsgId(), msg.getTopic(), msg.getTags(), content);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        broadcastConsumer.start();
        log.info("广播模式消费者启动成功");
    }

    /**
     * 初始化集群模式消费者
     * 集群模式下，同一个消费者组内的消费者平均分摊消息
     */
    private void initClusteringConsumer() throws MQClientException {
        clusteringConsumer = new DefaultMQPushConsumer(RocketMQConstant.CONSUMER_GROUP_NORMAL + "-clustering");
        clusteringConsumer.setNamesrvAddr(nameServer);
        clusteringConsumer.subscribe(RocketMQConstant.TOPIC_NORMAL, "*");

        // 设置为集群模式（默认）
        clusteringConsumer.setMessageModel(MessageModel.CLUSTERING);

        // 从最新的消息开始消费
        clusteringConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        // 设置每次拉取的最大消息数
        clusteringConsumer.setPullBatchSize(10);

        // 设置每次消费的最大消息数
        clusteringConsumer.setConsumeMessageBatchMaxSize(1);

        clusteringConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                    log.info("集群模式消费者接收到消息: msgId={}, topic={}, tags={}, content={}",
                            msg.getMsgId(), msg.getTopic(), msg.getTags(), content);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        clusteringConsumer.start();
        log.info("集群模式消费者启动成功");
    }

    /**
     * 初始化重试机制消费者
     * 演示消息重试机制
     */
    private void initRetryConsumer() throws MQClientException {
        retryConsumer = new DefaultMQPushConsumer(RocketMQConstant.CONSUMER_GROUP_NORMAL + "-retry");
        retryConsumer.setNamesrvAddr(nameServer);
        retryConsumer.subscribe(RocketMQConstant.TOPIC_NORMAL, "*");

        // 设置消费失败的最大重试次数（默认16次）
        retryConsumer.setMaxReconsumeTimes(3);

        retryConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                    log.info("重试机制消费者接收到消息: msgId={}, topic={}, tags={}, 重试次数={}, content={}",
                            msg.getMsgId(), msg.getTopic(), msg.getTags(), msg.getReconsumeTimes(), content);
                    // 模拟消费失败的情况
                    if (content.contains("error") || content.contains("fail")) {
                        log.warn("消息内容包含错误关键字，将进行重试");
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    // 如果重试次数达到2次，仍然失败，则记录日志但不再重试
                    if (msg.getReconsumeTimes() >= 2) {
                        log.error("消息重试次数已达到上限，记录失败日志: {}", content);
                        // 实际应用中，可以将失败消息存入数据库或死信队列
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        retryConsumer.start();
        log.info("重试机制消费者启动成功");
    }

    @PreDestroy
    public void destroy() {
        if (broadcastConsumer != null) {
            broadcastConsumer.shutdown();
            log.info("广播模式消费者已关闭");
        }

        if (clusteringConsumer != null) {
            clusteringConsumer.shutdown();
            log.info("集群模式消费者已关闭");
        }

        if (retryConsumer != null) {
            retryConsumer.shutdown();
            log.info("重试机制消费者已关闭");
        }
    }
}
